#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
爆仓数据处理器
处理实时爆仓数据，计算不同时间窗口的统计信息
"""

import asyncio
import time
from collections import defaultdict, deque
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from loguru import logger
import json

@dataclass
class LiquidationRecord:
    """爆仓记录数据结构"""
    exchange_name: str
    base_coin: str
    side: str  # buy/sell
    pos_side: str  # long/short
    volume: float
    amount: float
    price: float
    avg_price: float
    trade_turnover: float
    timestamp: int
    contract_code: str = ""
    exchange_type: str = ""
    contract_type: str = ""

@dataclass
class TimeWindowStats:
    """时间窗口统计数据"""
    long_volume: float = 0.0
    long_amount: float = 0.0
    long_count: int = 0
    short_volume: float = 0.0
    short_amount: float = 0.0
    short_count: int = 0
    total_volume: float = 0.0
    total_amount: float = 0.0
    total_count: int = 0
    last_update: int = 0

class LiquidationDataProcessor:
    """爆仓数据处理器"""
    
    def __init__(self):
        # 时间窗口配置(秒)
        self.time_windows = {
            '1m': 60,
            '3m': 180,
            '5m': 300,
            '10m': 600,
            '15m': 900,
            '30m': 1800
        }
        
        # 存储每个币种的原始数据 {coin: deque[LiquidationRecord]}
        self.raw_data: Dict[str, deque] = defaultdict(lambda: deque(maxlen=10000))
        
        # 存储统计数据 {coin: {window: TimeWindowStats}}
        self.stats_data: Dict[str, Dict[str, TimeWindowStats]] = defaultdict(
            lambda: {window: TimeWindowStats() for window in self.time_windows.keys()}
        )
        
        # 数据锁
        self.data_lock = asyncio.Lock()
        
        # 后台任务
        self._cleanup_task = None
        self._recompute_task = None
        self._recompute_interval = 1  # 秒
        
    async def start_background_tasks(self):
        """启动后台任务（清理与定期重算统计）"""
        if self._cleanup_task is None:
            self._cleanup_task = asyncio.create_task(self._cleanup_expired_data())
        if self._recompute_task is None:
            self._recompute_task = asyncio.create_task(self._recompute_stats_loop())
            
    async def stop_background_tasks(self):
        """停止后台任务"""
        for task_attr in ("_cleanup_task", "_recompute_task"):
            task = getattr(self, task_attr)
            if task:
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
                setattr(self, task_attr, None)
        
    # 兼容旧调用名称
    async def start_cleanup_task(self):
        await self.start_background_tasks()
    async def stop_cleanup_task(self):
        await self.stop_background_tasks()
        
    async def process_liquidation_data(self, data_list: List[Dict[str, Any]]):
        """处理爆仓数据列表"""
        async with self.data_lock:
            for data in data_list:
                try:
                    record = self._parse_liquidation_record(data)
                    if record:
                        await self._add_record(record)
                except Exception as e:
                    logger.error(f"处理爆仓数据失败: {e}, 数据: {data}")
                    
    def _parse_liquidation_record(self, data: Dict[str, Any]) -> Optional[LiquidationRecord]:
        """解析爆仓记录并规范化字段（时间戳单位、金额口径）"""
        try:
            # 规范时间戳为毫秒
            raw_ts = data.get('ts', 0)
            ts = int(raw_ts) if raw_ts is not None else 0
            if ts and ts < 1_000_000_000_000:  # 小于 1e12 认为是秒
                ts *= 1000
            
            trade_turnover = data.get('tradeTurnover')
            amount = data.get('amount')
            # 统一金额口径：优先使用 tradeTurnover；若为空则回退到 amount
            trade_turnover_val = float(trade_turnover) if trade_turnover is not None else float(amount or 0)
            amount_val = float(amount or 0)
            
            return LiquidationRecord(
                exchange_name=data.get('exchangeName', ''),
                base_coin=data.get('baseCoin', ''),
                side=data.get('side', ''),
                pos_side=data.get('posSide', ''),
                volume=float(data.get('volume', 0)),
                amount=amount_val,
                price=float(data.get('price', 0)),
                avg_price=float(data.get('avgPrice', 0)),
                trade_turnover=trade_turnover_val,
                timestamp=ts,
                contract_code=data.get('contractCode', ''),
                exchange_type=data.get('exchangeType', ''),
                contract_type=data.get('contractType', '')
            )
        except (ValueError, TypeError) as e:
            logger.error(f"解析爆仓记录失败: {e}, 数据: {data}")
            return None
            
    async def _add_record(self, record: LiquidationRecord):
        """添加爆仓记录并更新统计（立即重算一次）"""
        coin = record.base_coin
        
        # 添加到原始数据
        self.raw_data[coin].append(record)
        
        # 立即重算该币种所有窗口，确保前端及时看到衰减后的数据
        await self._recompute_coin_stats(coin)
        
        logger.debug(f"已处理 {coin} 爆仓数据: {record.pos_side} {record.volume}")
        
    def _calculate_window_stats(self, coin: str, cutoff_time: int) -> TimeWindowStats:
        """计算指定时间窗口的统计数据（基于当前 raw_data）"""
        stats = TimeWindowStats()
        
        records = self.raw_data.get(coin, [])
        for record in records:
            if record.timestamp >= cutoff_time:
                # 更新统计数据
                if record.pos_side and record.pos_side.lower() == 'long':
                    stats.long_volume += record.volume
                    stats.long_amount += record.trade_turnover
                    stats.long_count += 1
                elif record.pos_side and record.pos_side.lower() == 'short':
                    stats.short_volume += record.volume
                    stats.short_amount += record.trade_turnover
                    stats.short_count += 1
                
                stats.total_volume += record.volume
                stats.total_amount += record.trade_turnover
                stats.total_count += 1
                
                if record.timestamp > stats.last_update:
                    stats.last_update = record.timestamp
        
        return stats
        
    def _get_coin_stats_unlocked(self, coin: str) -> Optional[Dict[str, Any]]:
        """获取指定币种的统计数据（无锁版本）。读取前先按当前时间重算一遍，确保窗口衰减。"""
        if coin not in self.stats_data and coin not in self.raw_data:
            return None
        
        # 读取前重算
        current_time = int(time.time() * 1000)
        for window_name, window_seconds in self.time_windows.items():
            cutoff = current_time - window_seconds * 1000
            self.stats_data[coin][window_name] = self._calculate_window_stats(coin, cutoff)
            
        result = {
            'coin': coin,
            'timeWindows': {},
            'lastUpdate': 0
        }
        
        for window_name, stats in self.stats_data[coin].items():
            result['timeWindows'][window_name] = {
                'long': {
                    'volume': round(stats.long_volume, 6),
                    'amount': round(stats.long_amount, 2),
                    'count': stats.long_count
                },
                'short': {
                    'volume': round(stats.short_volume, 6),
                    'amount': round(stats.short_amount, 2),
                    'count': stats.short_count
                },
                'total': {
                    'volume': round(stats.total_volume, 6),
                    'amount': round(stats.total_amount, 2),
                    'count': stats.total_count
                }
            }
            
            if stats.last_update > result['lastUpdate']:
                result['lastUpdate'] = stats.last_update
                
        return result
    
    async def get_coin_stats(self, coin: str) -> Optional[Dict[str, Any]]:
        """获取指定币种的统计数据"""
        async with self.data_lock:
            return self._get_coin_stats_unlocked(coin)
            
    async def get_all_coins_stats(self) -> List[Dict[str, Any]]:
        """获取所有币种的统计数据（读取时按当前时间重算）"""
        async with self.data_lock:
            results = []
            
            # 遍历 raw_data 的键，确保包含新币种
            for coin in set(list(self.raw_data.keys()) + list(self.stats_data.keys())):
                coin_stats = self._get_coin_stats_unlocked(coin)
                if coin_stats:
                    results.append(coin_stats)
                    
            # 按总交易量排序
            results.sort(
                key=lambda x: x['timeWindows'].get('5m', {}).get('total', {}).get('amount', 0), 
                reverse=True
            )
            
            return results
            
    async def get_top_coins(self, limit: int = 50, window: str = '5m') -> List[Dict[str, Any]]:
        """获取爆仓量最大的币种"""
        all_stats = await self.get_all_coins_stats()
        
        # 按指定时间窗口的总交易量排序
        sorted_stats = sorted(
            all_stats,
            key=lambda x: x['timeWindows'].get(window, {}).get('total', {}).get('amount', 0),
            reverse=True
        )
        
        return sorted_stats[:limit]
        
    def get_total_records_count(self) -> int:
        """获取总记录数"""
        return sum(len(records) for records in self.raw_data.values())
        
    async def _cleanup_expired_data(self):
        """定期清理过期数据"""
        while True:
            try:
                await asyncio.sleep(60)  # 每分钟清理一次
                
                async with self.data_lock:
                    current_time = int(time.time() * 1000)
                    max_window_ms = max(self.time_windows.values()) * 1000
                    cutoff_time = current_time - max_window_ms - 60000  # 多保留1分钟
                    
                    # 清理过期的原始数据
                    for coin, records in self.raw_data.items():
                        # 从左侧移除过期记录
                        while records and records[0].timestamp < cutoff_time:
                            records.popleft()
                    
                    # 清理完成后重算一次所有币种的统计
                    for coin in self.raw_data.keys():
                        for window_name, seconds in self.time_windows.items():
                            cutoff = current_time - seconds * 1000
                            self.stats_data[coin][window_name] = self._calculate_window_stats(coin, cutoff)
                    
                    logger.debug(f"数据清理完成，当前总记录数: {self.get_total_records_count()}")
                    
            except Exception as e:
                logger.error(f"数据清理异常: {e}")
                
    async def _recompute_stats_loop(self):
        """后台定期重算所有币种的窗口统计，保证数值随时间衰减"""
        while True:
            try:
                await asyncio.sleep(self._recompute_interval)
                async with self.data_lock:
                    await self._recompute_all()
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"定期重算异常: {e}")
                
    async def _recompute_all(self):
        current_time = int(time.time() * 1000)
        for coin in self.raw_data.keys():
            for window_name, seconds in self.time_windows.items():
                cutoff = current_time - seconds * 1000
                self.stats_data[coin][window_name] = self._calculate_window_stats(coin, cutoff)
    
    async def _recompute_coin_stats(self, coin: str):
        current_time = int(time.time() * 1000)
        for window_name, seconds in self.time_windows.items():
            cutoff = current_time - seconds * 1000
            self.stats_data[coin][window_name] = self._calculate_window_stats(coin, cutoff)
        
    async def get_liquidation_records(self, coin: str, limit: int = 100, min_amount: float = 0) -> List[Dict[str, Any]]:
        """获取指定币种的逐笔爆仓记录"""
        async with self.data_lock:
            if coin not in self.raw_data:
                return []
            
            records = list(self.raw_data[coin])
            
            # 按时间戳倒序排列（最新的在前）
            records.sort(key=lambda x: x.timestamp, reverse=True)
            
            # 筛选最小金额
            if min_amount > 0:
                records = [r for r in records if r.amount >= min_amount]
            
            # 限制数量
            if limit > 0:
                records = records[:limit]
            
            # 转换为字典格式
            result = []
            for record in records:
                result.append({
                    'timestamp': record.timestamp,
                    'exchange': record.exchange_name,
                    'side': 'long' if record.pos_side.lower() == 'long' else 'short',
'amount': record.trade_turnover,  # 与聚合口径一致
                    'price': record.price,
                    'volume': record.volume,
                    'avg_price': record.avg_price,
                    'trade_turnover': record.trade_turnover
                })
            
            return result
    
    async def get_system_stats(self) -> Dict[str, Any]:
        """获取系统统计信息"""
        async with self.data_lock:
            return {
                'total_coins': len(self.raw_data),
                'total_records': self.get_total_records_count(),
                'time_windows': list(self.time_windows.keys()),
                'memory_usage': {
                    coin: len(records) for coin, records in self.raw_data.items()
                }
            }